package ml

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{StringIndexer, StringIndexerModel, VectorAssembler}
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
  *
  */
object IRISTree {
  def main(args: Array[String]): Unit = {
    //监督学习如何体现
    //训练集 ，测试集如何使用
    //成功率

    //1.初始化是sparkSession
    val spark: SparkSession = SparkSession.builder().appName("iris").master("local[*]").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")

    import spark.implicits._
    //2.读取数据
    val dataSource: DataFrame = spark.read.csv("file:///D:\\大数据\\学期文档\\项目\\03挖掘型标签\\数据集\\iris_tree.csv")
      .toDF("Sepal_Length", "Sepal_Width", "Petal_Length", "Petal_Width", "Species")
      .select(
        'Sepal_Length cast (DoubleType), //字符串无法直接计算，需要将其转化为DoubleType
        'Sepal_Width cast (DoubleType),
        'Petal_Length cast (DoubleType),
        'Petal_Width cast (DoubleType),
        'Species
      )
    //3,标签处理，(将最终的标签转化成数字数据不能直接使用)
    val SpeciesToLabel: StringIndexerModel = new StringIndexer()
      .setInputCol("Species")
      .setOutputCol("label")
      .fit(dataSource)

    //4.将特征数据转化为向量
    val features: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("Sepal_Length", "Sepal_Width", "Petal_Length", "Petal_Width"))
      .setOutputCol("features")

    //5.初始化决策树,进行分类
    val decisionTree: DecisionTreeClassifier = new DecisionTreeClassifier() //创建决策树对象
      .setFeaturesCol("features") //设置特征列
      .setPredictionCol("prediction") //设置预测后的列名
      .setMaxDepth(4) //设置深度

    //6.构建一个pipLine，将3 、4、5步骤连起来
    val pipeline: Pipeline = new Pipeline()
      .setStages(Array(SpeciesToLabel, features, decisionTree))

    //7.将原始数据拆分成训练集和测试集
    val Array(trainDatas, testDatas): Array[Dataset[Row]] = dataSource.randomSplit(Array(0.8, 0.2))

    //8 使用pipline对训练集进行训练，训练完使用测试集进行测试
    //使用训练数据进行训练 得到一个模型
    val trainModel: PipelineModel = pipeline.fit(trainDatas)
    //使用测试集进行测试
    val testDF: DataFrame = trainModel.transform(testDatas)
    //testDF.show()

    //9 查看分类成功的百分比
    val evaluator: MulticlassClassificationEvaluator = new MulticlassClassificationEvaluator() //多类别评估器
      .setLabelCol("label") //设置原始数据中自带的label
      .setPredictionCol("prediction") //设置根据数据计算出来的结果
    val Score: Double = evaluator.evaluate(testDF) //将测试数据集中自带的结果与计算出来的结果进行对比
    println(Score)
    //10 看一看这个决策树过程
    val model: DecisionTreeClassificationModel = trainModel.stages(2).asInstanceOf[DecisionTreeClassificationModel]
    println(model.toDebugString)
  }
}
